package com.bw.app.functions;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.bw.bean.TableProcess;
import com.bw.common.GmallConfig;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.*;

public class TableProcessFunction extends BroadcastProcessFunction<JSONObject, String, JSONObject> {

    private Connection connection;// ctrl+alt+f 变量提升
    private MapStateDescriptor<String, TableProcess> mapStateDescriptor;
    private OutputTag<JSONObject> hbaseTag;
    private PreparedStatement preparedStatement = null;


    public TableProcessFunction(MapStateDescriptor<String, TableProcess> mapStateDescriptor, OutputTag<JSONObject> hbaseTag) {
        this.mapStateDescriptor = mapStateDescriptor;
        this.hbaseTag = hbaseTag;
    }


    @Override
    public void open(Configuration parameters) throws Exception {
        // 做Hbase链接
        Class.forName(GmallConfig.PHOENIX_DRIVER);
        // 一个并行度只有一个connection
        connection = DriverManager.getConnection(GmallConfig.PHOENIX_SERVER);
        connection.setAutoCommit(true);
    }

    // 主流
    @Override
    public void processElement(JSONObject jsonObject, ReadOnlyContext readOnlyContext, Collector<JSONObject> collector) throws Exception {
        // 1.取出广播状态
        ReadOnlyBroadcastState<String, TableProcess> broadcastState = readOnlyContext.getBroadcastState(mapStateDescriptor);
        String key = jsonObject.getString("tableName") + "-" + jsonObject.getString("type");
        TableProcess tableProcess = broadcastState.get(key);// tableProcess有可能为null
        if (tableProcess != null) {
            // 2.过滤 字段
            JSONObject after = jsonObject.getJSONObject("after");
            filter(after, tableProcess.getSinkColumns());

            // 3.分流
            // kafka主流  hbase侧流
            String sinkType = tableProcess.getSinkType();
            String sinkTable = tableProcess.getSinkTable();
            // 把sinkTable放到这里面，后面才能知道输出到哪张表或者哪个话题下
            jsonObject.put("sinkTable", sinkTable);
            if (TableProcess.SINK_TYPE_KAFKA.equals(sinkType)) {
                collector.collect(jsonObject);
            } else if (TableProcess.SINK_TYPE_HBASE.equals(sinkType)) {
                readOnlyContext.output(hbaseTag, jsonObject);
            }
        } else {
            System.out.println("该组合不存在");
        }


    }


    // 广播流
    @Override
    public void processBroadcastElement(String value, Context context, Collector<JSONObject> collector) throws Exception {
        // 1. 转成JavaBean对象
        String data = JSONObject.parseObject(value).getString("after");
        TableProcess tableProcess = JSON.parseObject(data, TableProcess.class);
        // 2. 动态建表
        if (TableProcess.SINK_TYPE_HBASE.equals(tableProcess.getSinkType())) {
            // create table if not exist db.table_name (id primary key,name varchar) egine=innodb
            checkTable(tableProcess.getSinkTable(), tableProcess.getSinkColumns(), tableProcess.getSinkPk(), tableProcess.getSinkExtend());
        }
        // 3. 广播状态
        BroadcastState<String, TableProcess> broadcastState = context.getBroadcastState(mapStateDescriptor);
        // base_tardemark-insert
        String key = tableProcess.getSourceTable() + "-" + tableProcess.getOperateType();
        // 广播状态
        broadcastState.put(key, tableProcess);
    }

    // 建表语句
    //create table if not exist db.table_name (id varchar primary key,name varchar) egine=innodb
    private void checkTable(String sinkTable, String sinkColumns, String sinkPk, String sinkExtend) {
        if (sinkExtend == null) {
            sinkExtend = "";
        }
        if (sinkPk == null) {
            sinkPk = "id";
        }
        StringBuffer createSql = new StringBuffer()
                .append("create table if not exists ")
                .append(GmallConfig.HBASE_SCHEMA)
                .append(".")
                .append(sinkTable)
                .append(" (");
        // 切分字段
        String[] fields = sinkColumns.split(",");
        for (int i = 0; i < fields.length; i++) {
            String field = fields[i];
            // 如果是主键
            if ("id".equals(field)) {
                createSql.append(sinkPk).append(" varchar primary key ");
            } else {
                createSql.append(field).append(" varchar ");
            }
            // 拼接逗号
            if (i != fields.length - 1) {
                createSql.append(",");
            }
        }
        createSql.append(")").append(sinkExtend);
        // 打印SQL
        System.out.println("createSQL = " + createSql);
        // 执行SQL
        try {
            preparedStatement = connection.prepareStatement(createSql.toString());
            preparedStatement.executeUpdate();
        } catch (SQLException throwables) {
            throwables.printStackTrace();
        } finally {
            if (preparedStatement != null) {
                try {
                    preparedStatement.close();
                } catch (SQLException throwables) {
                    throwables.printStackTrace();
                }
            }
        }

    }

    // 过滤
    private void filter(JSONObject after, String sinkColumns) {
        // after {“id":1,"tm_name":"bw","logo_url":"/static/1.png"}
        // sinkColumns  id,tm_name
        List<String> columns = Arrays.asList(sinkColumns.split(",")); // [id,tm_name]
        // ![id,tm_name] contains logo_url
        after.entrySet().removeIf(next -> !columns.contains(next.getKey()));

//       原代码
//        String columns = Arrays.toString(sinkColumns.split(",")); // [id,tm_name]
//        Iterator<Map.Entry<String, Object>> iterator = after.entrySet().iterator();
//        while (iterator.hasNext()) {
//            Map.Entry<String, Object> next = iterator.next();
//            if(!columns.contains(next.getKey())){// ![id,tm_name] contains logo_url
//                iterator.remove();
//            }
//        }
    }
}
